/*
 * This file is a part of qloudgen-esb. 
 * You can redistribute qloudgen-esb and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-esb is distributed WITHOUT ANY WARRANTY. 
 * 
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/.
 *  
 * Copyright (C) 2014 Qloudgen Technology
*/

package com.qloudgen.csb.client;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;

import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;

import org.apache.log4j.Logger;

public class KfkConsumer implements Runnable {
	private String zkConnect;
	private String id;

	private int count = 0;
    private ConsumerConnector consumer;
    private ConsumerIterator<byte[], byte[]> it;
    
    private String topic;
    private String groupId;
    private int bufferSize;
    
    private static Logger LOG = Logger.getLogger(KfkConsumer.class.getName());

    public KfkConsumer(String topic, String groupId,
    		String zkConnect, int bufferSize, String id) {
    	
		this.zkConnect = zkConnect;
		if (id == null)
			this.id = UUID.randomUUID().toString().replace("-", "");
		else
			this.id = id;

		this.topic = topic;
		this.groupId = groupId;
		
		ConsumerConfig config = createConsumerConfig();
    	consumer = kafka.consumer.Consumer.
            	createJavaConsumerConnector(config);
        Map<String, Integer> topicCountMap = 
        		new HashMap<String, Integer>();
        topicCountMap.put(this.topic, new Integer(1));
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = 
        		consumer.createMessageStreams(topicCountMap);
        KafkaStream<byte[], byte[]> stream =  
        		consumerMap.get(this.topic).get(0);
        it = stream.iterator();
        LOG.info(this.id + ": consumer created...");
    }
        
	KfkMessageListener listener = null;
	public void setMessageListener(KfkMessageListener listener) {
		this.listener = listener;
	}

    public void run() {        
        count = 0;
        while (it.hasNext()) {
        	MessageAndMetadata<byte[], byte[]> mam = it.next();
        	byte[] message = mam.message();
            long offset = mam.offset();
            listener.onMessage(new String(message), offset);
        	count++;
        	if (count >= bufferSize) {
        		consumer.commitOffsets();
        		count = 0;
        	}
        }        	        	
        LOG.info(id + ": consumer stopped...");
        consumer.shutdown();
    } 
    
    private ConsumerConfig createConsumerConfig() {
    	Properties props = new Properties();
    	FileInputStream fis = null;
		try {
			fis = new FileInputStream("conf/csbconsumer.properties");
			props.load(fis);
		} catch (IOException e) {			
		} finally {
			if (fis != null)
				try {
					fis.close();
				} catch (IOException e) {}
		}
		
        props.put("zookeeper.connect", zkConnect);
        props.put("group.id", groupId);
        props.put("auto.commit.enable", "false");
        props.put("auto.offset.reset", "largest");
        props.put("consumer.timeout.ms", "-1");
        props.put("client.id", KfkConsumer.class.getName());
        
        return new ConsumerConfig(props);
    }  
}

